package com.nightsoul.hadoop1.test.format;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.lib.MultipleOutputs;
import org.apache.hadoop.mapred.lib.NullOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.nightsoul.hadoop1.test.OldJobBuilder;
import com.nightsoul.hadoop1.test.junit.NcdcRecordParser;

public class PartitionByStationUsingMultipleOutputs extends Configured implements Tool {
	/**
	 * StationMapper 将stationId从记录中抽取出来并将它作为键，这样可以将每一个气象站的
	 * 数据切分到同一个分区，默认使用的Partitioner为HashPartitioner
	 * HashPartitioner可以根据集群的容量来决定分区的个数，因为reducer的任务槽越多，任务越快完成，
	 * 并且还可以适应任意数量的分区，能够较好的保证Partitioner之间是比较均匀的。
	 * 所以在使用HashPartitioner时reduce任务数量是不固定的。
	 * 
	 * 使用HashPartitioner每个分区上将包含多个气象站的数据，如果需要实现每一个气象站一个输出文件，我们就需要让每个reducer写多个文件，
	 * 使用MultipleOutputs或MultipleOutputFormat
	 * @author zj
	 *
	 */
	static class StationMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
		private NcdcRecordParser parser = new NcdcRecordParser();
		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			
			parser.parse(value);
			output.collect(new Text(parser.getStationId()), value);
		}
		
	}
	
	/**
	 * 流程处理进入到reduce之前，已经分好区了，MultipleOutputs只是增加了输出
	 * @author zj
	 *
	 */
	static class MultipleOutputsReducer extends MapReduceBase implements Reducer<Text, Text, NullWritable, Text> {
		private MultipleOutputs multipleOutputs;
		
		@Override
		public void configure(JobConf job) {
			super.configure(job);
			this.multipleOutputs = new MultipleOutputs(job);
		}
		
		@SuppressWarnings("unchecked")
		@Override
		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<NullWritable, Text> output, Reporter reporter)
				throws IOException {
			OutputCollector<NullWritable, Text> collector = multipleOutputs.getCollector("station", 
					key.toString().replace("-", ""), reporter);
			while(values.hasNext()) {
				collector.collect(NullWritable.get(), values.next());
			}
		}
		
		@Override
		public void close() throws IOException {
			super.close();
			this.multipleOutputs.close();
		}
	}
	
	
	
	@Override
	public int run(String[] args) throws Exception {
		JobConf conf = OldJobBuilder.parseInputAndOutput(this, getConf(), args);
		if(conf==null) {
			return -1;
		}
		
		conf.setMapperClass(StationMapper.class);
		conf.setMapOutputKeyClass(Text.class);
		conf.setReducerClass(MultipleOutputsReducer.class);
		conf.setOutputKeyClass(NullWritable.class);
		conf.setOutputFormat(NullOutputFormat.class);//supress empty part file
		
		MultipleOutputs.addMultiNamedOutput(conf, "station", TextOutputFormat.class, NullWritable.class, Text.class);;
		
		JobClient.runJob(conf);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		int exitCode = ToolRunner.run(new PartitionByStationUsingMultipleOutputs(), args);
		System.exit(exitCode);
	}
}
